package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import scala.Tuple2;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SAN041;
import com.surfilter.massdata.spark.model.DnsStatConsts;
import com.surfilter.massdata.spark.model.StructIP;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.IPDatabase;

/**
 * 运营商网站访问量
 * 各运营商访问量接入网站访问量变化统计（要求是省内接入网站，即dip在BR2002中存在）
 */
public class WebSiteVisitsOperatorTask_dayTask implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(WebSiteVisitsOperatorTask_dayTask.class);
	private static final long serialVersionUID = 1L;
	private String outName;
	
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		log.info("==============WebSiteVisitsOperatorTask begin====================");
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try{
			String dayStr = commandMap.get("-d");
	        Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
		DataFrame dataSource = dataFrames.get("dws_hour_san041");
		String sql="select buss_value,sum(count_value) as count from dws_hour_san041_temp where BUSS_TYPE='WEBSITE_COUNT_OPERATORS' and sta_range=0 group by buss_value";
		DataFrame df_sa041 = dataSource.sqlContext().sql(sql);
		JavaRDD<Map<String, String>> mapRDD = df_sa041.toJavaRDD().map(new Function<Row,Map<String,String>>(){

			@Override
			public Map<String, String> call(Row row) throws Exception {
				Map<String, String> map=new HashMap<String,String>();
				String buss_value=row.getAs("buss_value").toString();
				 String count=row.getAs("count").toString();
				map.put("buss_value", buss_value);
				map.put("count",count);
				
				 return  map;
			}
			
		}); 
        JavaRDD<SAN041> javaRDD = createRdd(date, mapRDD);
    	DataFrame df = dataSource.sqlContext().createDataFrame(javaRDD, SAN041.class);
 
		outList.add(new OutQueueEntity(outName, df));
		
		CommonUtils.deleteTaskTableData("SAN041", date, "WEBSITE_COUNT_OPERATORS", 0, "day");
	}catch(Exception e){
		if(log.isErrorEnabled()){
			log.error(e.getMessage(),e);
		}
	}
	
	long end=System.currentTimeMillis();
	double min=(end-start)*1.0/(1000*60);
	System.out.println("WebSiteVisitsOperatorTask_New运营商网站访问量天任务:exectime: "+min+" min............");
	return outList;
}

private JavaRDD<SAN041> createRdd(final Date date,JavaRDD<Map<String, String>> mapRDD) {
	JavaRDD<SAN041> javaRDD = mapRDD.mapPartitions(new FlatMapFunction<Iterator<Map<String,String>>,SAN041>() {

		@Override
		public Iterable<SAN041> call(Iterator<Map<String,String>> it){
			List<SAN041> list = new ArrayList<SAN041>();
			try{
				while(it.hasNext()){
					 Map<String, String> map = it.next();
					String operators_id = map.get("buss_value").toString();
					long count=Long.parseLong(map.get("count").toString());
					if(StringUtils.isNotBlank(operators_id)){
		        		SAN041 sa = new SAN041();
		        		sa.setYear(DateUtil.getCurrentYear(date));
		        		sa.setHalf_year(DateUtil.getHalfYear(date));
		        		sa.setQuarter(DateUtil.getQuarter(date));
		        		sa.setMonth(DateUtil.getCurrentMonth(date));
		        		sa.setWeek(DateUtil.getCurrentWeek(date));
		        		sa.setDay(DateUtil.getCurrentDay(date));
		        		sa.setHour(null);
		        		sa.setBuss_type("WEBSITE_COUNT_OPERATORS");
		        		sa.setBuss_value(operators_id);
		        		sa.setCount_value(count);
		        		sa.setSta_range(0);
		        		list.add(sa);
		        	}
				}
			}catch(Exception e){
				if(log.isErrorEnabled()){
					log.error(e.getMessage(),e);
				}
			}
			return list;
		}
	});
	return javaRDD;
}
}